package com.gome.han.bigdata.spark.core.rdd.operation.transformation

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author Hanpeng
 * @date 2021/1/13 13:28
 * @description:
 */
object GroupByOperation3 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)

    // TODO 算子 - groupBy
    val rdd = sc.textFile("in/apache.log")

   /* val timeRDD: RDD[(String, Iterable[(String, Int)])] = rdd.map(
      line => {
        val datas = line.split(" ")
        val time = datas(3)
        //time.substring(0, )
        val sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
        val date: Date = sdf.parse(time)
        val sdf1 = new SimpleDateFormat("HH")
        val hour: String = sdf1.format(date)
        (hour, 1)
      }
    ).groupBy(_._1)*/
   val lineRdd: RDD[(String, Int)] = rdd.map((line: String) => {
     val datas = line.split(" ")
     val time = datas(3)
     //time.substring(0, )
     val sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
     val date: Date = sdf.parse(time)
     val sdf1 = new SimpleDateFormat("HH")
     val hour: String = sdf1.format(date)
     (hour, 1)
   })
    // _._1表示每个元素的第一个元素(对应数组0) 每个元素都是一个Tuple
    val timeRdd: RDD[(String, Iterable[(String, Int)])] = lineRdd.groupBy(t=>{
      println(t.getClass);// scala.Tuple2
      t._1
    })
    timeRdd.map{
      case ( hour, iter ) => {
        (hour, iter.size)
      }
    }.collect.foreach(println)
    timeRdd.map(num =>{
       num match {
         case (str, dd) =>{
           (str, dd.size)
         }
       }
    })

    sc.stop()
  }
}
